package cn.doitedu.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/5/11
 * @Desc: 学大数据，上多易教育
 *   各品类累积到当前，支付总额最大的前10种商品及其支付总额（只统计已支付的）
 **/
public class Job04_BrandTopnPayProduct {
    public static void main(String[] args) {

        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);


        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        TableConfig config = tenv.getConfig();
        config.set("table.exec.source.idle-timeout","1 ms");

        // 开启窗口的提前触发器
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true);
        // 把触发器的触发周期设置为1秒
        config.getConfiguration().setString("table.exec.emit.early-fire.delay", "1s");


        // 建表，映射订单主表
        tenv.executeSql(
                " create table oms_order_cdc(                     "+
                        "    id bigint,                           "+  // 订单id
                        "    status int,                          "+  // 订单状态
                        "    modify_time timestamp(3),             "+  // 数据更新时间
                        "    payment_time timestamp(3),            "+  //支付时间
                        "    rt as modify_time,                   "+
                        "    watermark for rt as rt ,             "+
                        "    primary key(id) not enforced         "+
                        " ) WITH (                                "+
                        "     'connector' = 'mysql-cdc',          "+
                        "     'hostname' = 'doitedu',             "+
                        "     'port' = '3306',                    "+
                        "     'username' = 'root',                "+
                        "     'password' = 'root',                "+
                        "     'database-name' = 'realtimedw',     "+
                        "     'table-name' = 'oms_order'          "+
                        " )                                       "
        );

        // 建表，映射订单详情表
        tenv.executeSql(
                " create table oms_order_item_cdc(                "+
                        "    id bigint,                           "+  // 订单id
                        "    order_id bigint,                     "+  // 订单id
                        "    product_id bigint,                   "+  // 商品id
                        "    product_brand string,                "+  // 商品的品牌
                        "    real_amount decimal(10,2),           "+  // 实付金额
                        "    primary key(id) not enforced         "+
                        " ) WITH (                                "+
                        "     'connector' = 'mysql-cdc',          "+
                        "     'hostname' = 'doitedu',             "+
                        "     'port' = '3306',                    "+
                        "     'username' = 'root',                "+
                        "     'password' = 'root',                "+
                        "     'database-name' = 'realtimedw',     "+
                        "     'table-name' = 'oms_order_item'     "+
                        " )                                       "
        );

        // sql
//        tenv.executeSql("with tmp as (\n" +
//                "    select\n" +
//                "       product_brand as brand,\n" +
//                "       product_id,\n" +
//                "       sum(real_amount) as real_amount\n" +
//                "    from oms_order_cdc o  \n" +
//                "    join oms_order_item_cdc i\n" +
//                "    on o.id = i.order_id\n" +
//                "    where status in (1,2,3) and date_format(payment_time,'yyyy-MM-dd') = current_date\n" +
//                "    group by \n" +
//                "       product_brand,\n" +
//                "       product_id\n" +
//                ")\n" +
//                "\n" +
//                "select\n" +
//                "       brand,\n" +
//                "       product_id,\n" +
//                "       real_amount,\n" +
//                "       rn\n" +
//                "from (\n" +
//                "select\n" +
//                "       brand,\n" +
//                "       product_id,\n" +
//                "       real_amount,\n" +
//                "       row_number() over(partition by brand order by real_amount desc ) as rn\n" +
//                "from tmp )\n" +
//                "where rn<=10").print();


        tenv.executeSql("with joined as (\n" +
                "    select\n" +
                "       i.product_brand,\n" +
                "       i.product_id,\n" +
                "       i.real_amount,\n" +
                "       o.rt \n" +
                "    from oms_order_cdc o  \n" +
                "    left join oms_order_item_cdc i\n" +
                "    on o.id = i.order_id\n" +
                "    where status in (1,2,3) and date_format(payment_time,'yyyy-MM-dd') = current_date   \n" +
                ")\n" +

                ",tmp1 as (\n" +
                "    select\n" +
                "         tumble_start(rt,interval '1' hour) as window_start,\n" +
                "         tumble_end(rt,interval '1' hour) as window_end,\n" +
                "         product_brand,\n" +
                "         product_id,\n" +
                "         sum(real_amount) as real_amount\n" +
                "    from joined \n" +
                "    group by \n" +
                "        tumble(rt,interval '1' hour),\n" +
                "        product_brand,\n" +
                "        product_id\n" +
                ")    \n" +
                "\n" +
                ",tmp2 as (\n" +
                "\n" +
                "select\n" +
                "    window_start,\n" +
                "    window_end,\n" +
                "    product_brand,\n" +
                "    product_id,\n" +
                "    real_amount,\n" +
                "    row_number() over(partition by window_start,window_end,product_brand order by real_amount desc ) as rn\n" +
                "from tmp1 \n" +
                ")\n" +
                "select * from tmp2 where rn<=10   \n"
                ).print();


    }
}
